
#pragma once;
#include <vector>
#include <iostream>
#include <assert.h>
#include <algorithm>
#include <string>
#include <stdio.h>
#include <string.h>
#include <time.h> 
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <mutex>
#include <thread>
#include <memory>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <pthread.h>
#include <condition_variable>
#include <signal.h>

using namespace std;

#define INF 0
#define DBG 1
#define ERR 2
#define DEFAULT_LOG_LEVEL INF

#define LOG(level, format, ...)                                                               \
	do                                                                                        \
	{                                                                                         \
		if (DEFAULT_LOG_LEVEL > level)                                                        \
			break;                                                                            \
		time_t t = time(NULL);                                                                \
		struct tm *lt = localtime(&t);                                                        \
		char buffer[64] = {0};                                                                \
		strftime(buffer, 63, "%H:%M:%S", lt);                                                 \
		fprintf(stdout, "[%p %s-%s:%d]" format "\n",(void*)pthread_self(), buffer, __FILE__, __LINE__, ##__VA_ARGS__); \
	} while (0)

#define ILOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DLOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)
#define BUFFER_DEFAULT_SIZE 1024
class Any
{
private:
	class holder
	{
	public:
		virtual ~holder() {}
		virtual const type_info& type() = 0;
		virtual holder* clone() = 0;
	};

	template<class T>
	class placeholders : public holder
	{
	public:
		placeholders(const T& val)
			:_val(val)
		{}
		virtual const type_info& type(){ return typeid(T);}
		virtual holder* clone() { return new placeholders(_val);}
	public: 
		T _val;

	};		

	// 基类指针存放子类对象，实现多态 	
	holder* _content;

public:

	Any()
		:_content(nullptr)
	{}
	// 使用模板函数来编写构造函数, 这样做的目的是为了获取存储值的数据类型
	template<class T>
	Any(const T& val)
	{
		_content = new placeholders<T>(val);	
	}

	Any(const Any& other)
	{
		_content = other._content == nullptr ? nullptr : other._content->clone();
	}

	// 返回保存数据对象的指针
	template<class T>
	T* get()
	{
		assert(typeid(T) == _content->type());
		return &((placeholders<T>*)_content)->_val;
	}

	~Any()
	{
		delete _content;
	}

	Any& Swap(Any& other)
	{
		swap(_content,other._content);
		return *this;
	}

	template<class T>
	Any& operator=(const T& val)
	{
		Any(val) .Swap(*this);
		return *this;
	}

	Any& operator=(const Any& other)
	{
		Any(other) .Swap(*this);
		return *this;
	}

};
class Buffer
{
private:
	std::vector<char> _buffer;

	uint64_t _read_index;  // 读偏移
	uint64_t _write_index; // 写偏移
public:
	// 构造函数:首先需要指定我们缓冲区的大小
	Buffer()
		: _buffer(BUFFER_DEFAULT_SIZE), _read_index(0), _write_index(0) {}

	// 使用迭代器直接获取地址比较……规范，既然如此，也就学来！
	// 返回初始位置的地址
	char *Begin() { return &*_buffer.begin(); }
	// 1.获取当前写位置地址
	char *Get_write_index() { return Begin() + _write_index; }
	// 6.获取当前读位置地址
	char *Get_read_index() { return Begin() + _read_index; }
	// 2.1：获取前沿空闲空间大小
	uint64_t Get_front_space() { return _read_index; }
	// 2.2:获取后沿空闲空间大小
	uint64_t Get_back_space() { return _buffer.size() - _write_index; }
	// 5.将写位置向后移动指定长度
	void Move_write_index(const uint64_t &len)
	{
		// 向后写的长度不能超过后沿剩余长度
		assert(len <= Get_back_space());
		_write_index += len;
	}
	// 7.获取可读数据大小
	uint64_t Size_of_read() { return _write_index - _read_index; }

	// 8.将读位置向后移动指定长度
	void Move_read_index(const uint64_t &len)
	{
		if (len == 0) return;
		assert(len <= Size_of_read());
		_read_index += len;
	}

	// ⒉确保可写空间足够(移动+扩容)获取前沿空闲空间大小，
	// 首先前沿的空间 + 后沿的空间必须大于写入数据的空间大小，如果小于则进行扩容
	void Ensure_write_space(const uint64_t &len)
	{
		// 如果后沿空间足够，直接返回，无需动作

		if (Get_back_space() >= len)
		{
			return;

		}
		// 如果后沿空间不足，前沿 + 后沿反而足够，则贪，将数据挪动到开头位置
		if (Get_back_space() + Get_front_space() >= len)
		{
			uint64_t read_size = Size_of_read();
			std::copy(Get_read_index(), Get_read_index() + read_size, Begin());
			// 拷贝结束更新偏移量
			_read_index = 0;
			_write_index = read_size;
		}
		// 如果没有足够的长度给予，则扩容
		else
		{
			_buffer.resize(len + _write_index);
		}
	}

	// 读取数据
	void Read(void *data, uint64_t len)
	{
		assert(len <= Size_of_read());
		std::copy(Get_read_index(), Get_read_index() + len, (char *)data);
	}
	// 读取数据并弹出数据
	void Read_Pop(void *data, uint64_t len)
	{
		Read(data, len);
		Move_read_index(len);
	}
	// 以String形式返回读取的数据
	std::string Read_as_string(uint64_t len)
	{
		assert(len <= Size_of_read());
		std::string str;
		str.resize(len);
		Read(&str[0], len);
		return str;
	}
	// 以String形式返回读取的数据,并弹出数据
	std::string Read_as_string_Pop(uint64_t len)
	{
		std::string ret = Read_as_string(len);
		Move_read_index(len);
		return ret;
	}

	// 写入数据
	void Write(const void *data, uint64_t len)
	{
		if (len == 0)
			return;

		Ensure_write_space(len);
		const char *d = (const char *)data;
		std::copy(d, d + len, Get_write_index());
	}

	// 写入数据后更新下标,推入数据
	void Write_push(const void *data, uint64_t len)
	{
		Write(data, len);
		Move_write_index(len);
	}

	// 以string类型进行写入
	void Write_as_string(const std::string &str)
	{
		Write(str.c_str(), str.size());
	}

	// 以string类型写入后放入数据
	void Write_as_string_push(const std::string &str)
	{
		Write(str.c_str(), str.size());
		Move_write_index(str.size());
	}

	// 以Buffer为类型进行写入数据
	void Write_as_buffer(Buffer &buf)
	{
		return Write(buf.Get_read_index(), buf.Size_of_read());
	}

	void Write_as_buffer_push(Buffer &buf)
	{
		cout<<"写入的字符串内容为："<<buf.Get_read_index()<<endl;
		Write_as_buffer(buf);
		Move_write_index(buf.Size_of_read());
	}

	char *FindCRLF()
	{
		char *res = (char *)memchr(Get_read_index(), '\n', Size_of_read());
		return res;
	}
	/*通常获取一行数据，这种情况针对是*/
	std::string GetLine()
	{
		char *pos = FindCRLF();
		if (pos == NULL)
		{
			return "";
		}
		// +1是为了把换行字符也取出来。
		return Read_as_string(pos - Get_read_index() + 1);
	}
	std::string GetLineAndPop()
	{
		std::string str = GetLine();
		Move_read_index(str.size());
		return str;
	}
	// 9.清理功能

	void Clean() { _read_index = _write_index = 0; }
};

// 编写一个套接字类，承担获取链接，发起链接，发送数据，接收数据的功能，其具体的实现思想不同于我们
// 分开实现服务端和客户端

#define MAX_LISTEN 1024
class Socket
{
private:
	int _socketfd;

public:
	Socket() : _socketfd(-1) {}
	Socket(int fd) : _socketfd(fd) {}
	~Socket() { Close(); }
	int FD() { return _socketfd; }
	// 创建套接字
	bool Creat_socket()
	{
		_socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
		if (_socketfd == -1)
		{
			ELOG("socket build FAILED!");
			return false;
		}
		return true;
	}

	// 绑定地址信息
	bool Bind(const std::string &ip, uint16_t port)
	{
		// 填充sockaddr_in结构体
		struct sockaddr_in local;
		local.sin_addr.s_addr = inet_addr(ip.c_str()); // inet_addr(ip.c_str())，将 ip 转换为网络字节序的 IP 地址
		local.sin_port = htons(port);				   // 绑定端口号 host_to_net_short端口转网络字节序
		local.sin_family = AF_INET;					   // 字段指定协议类型为 IPv4。
		int ret = bind(_socketfd, (sockaddr *)&local, sizeof(local));
		if (ret < 0)
		{
			ELOG("bind address FAILED!");
			return false;
		}
		return true;
	}

	// 开始监听,blokclog即同一时间的最大并发数,也就是等待连接队列的长度
	bool Listen(int blocklog = MAX_LISTEN)
	{
		int ret = listen(_socketfd, blocklog);
		if (ret < 0)
		{
			ELOG("listen FAILED!");
			return false;
		}
		return true;
	}

	// 向服务器发起连接,
	bool Connect(const std::string &ip, uint16_t port)
	{
		// 填写需要建立链接的目标主机的ip和端口号,调用connect函数
		// connect(int __fd, const sockaddr *__addr, socklen_t __len)
		struct sockaddr_in server_to_connect;
		server_to_connect.sin_family = AF_INET;
		server_to_connect.sin_port = htons(port);
		server_to_connect.sin_addr.s_addr = inet_addr(ip.c_str());

		socklen_t sockadd_len = sizeof(server_to_connect);

		int ret = connect(_socketfd, (sockaddr *)&server_to_connect, sockadd_len);

		if (ret < 0)
		{
			ELOG("connect FAILED !");
			return false;
		}
		return true;
	}

	// 获取新连接
	int Accept()
	{
		// 获取到来的链接
		struct sockaddr_in peer;
		socklen_t len = sizeof(peer);
		int accept_fd = accept(_socketfd, (sockaddr *)&peer, &len);
		if (accept_fd < 0)
		{
			DLOG("接受链接失败!");
			return -1;
		}
		else
			accept_fd;
	}

	// 接收数据,flag 默认为 0 即为阻塞操作
	ssize_t Recv(void *buf, size_t len, int flag = 0)
	{
		ssize_t ret = recv(_socketfd, buf, len, flag);
		if (ret <= 0)
		{
			if (errno == EAGAIN || errno == EINTR)
				return 0; // 表示没有收到数据
			ELOG("socket recv FAILED!");
			return -1;
		}
		return ret;
	}

	ssize_t No_block_recv(void *buf, size_t len)
	{
		return Recv(buf, len, MSG_DONTWAIT);
	}

	// 发送数据
	ssize_t Send(const void *buf, size_t len, int flag = 0)
	{
		ssize_t ret = send(_socketfd, buf, len, flag);
		if (ret < 0)
		{
			if (errno == EAGAIN || errno == EINTR)
				return 0; // 表示没有收到数据
			ELOG("socket Send FAILED!");
			return -1;
		}
		return ret;
	}
	// 设置为非阻塞发送
	ssize_t No_block_send(void *buf, size_t len)
	{
		if(len == 0) return 0;
		return Send(buf, len, MSG_DONTWAIT);
	}

	// 关闭套接亨
	void Close()
	{
		if (_socketfd != -1)
		{
			close(_socketfd);
			_socketfd = -1;
		}
	}
	// 创建一个服务端链接
	bool CreatServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flag = false)
	{
		// 1.创建套接字 2.绑定IP地址以及端口 3.开始进入监听状态 4.获取新链接 5.启用地址重用
		if (Creat_socket() == false)
			return false;
		if (flag)
			SetNoBlock();
		if (Bind(ip, port) == false)
			return false;
		if (Listen() == false)
			return false;
		ResuseAddress();

		return true;
	}

	// 创建一个客户端连接
	bool CreatClient(uint16_t port, const std::string &ip)
	{
		// 客户端只需要指定需要访问的指定服务器的端口以及IP即可
		if (Creat_socket() == false)
			return false;
		if (Connect(ip, port) == false)
			return false;

		return true;
	}

	// 开启端口重用
	void ResuseAddress()
	{
		int val = 1;
		setsockopt(_socketfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(int));
		val = 1;
		setsockopt(_socketfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, sizeof(int));
	}

	// 设置套接字的阻塞属性设置为非阻塞
	void SetNoBlock()				
	{
		// 先对文件当前的状态进行获取
		int flag = fcntl(_socketfd, F_GETFL);
		fcntl(_socketfd, F_SETFL, flag | O_NONBLOCK);
	}
};



// 通道模块
// 功能：对一个描述符进行监控事件管理
// 具体如何进行管理，则需要更加清晰的了解其实现的功能
class Poller;
class Eventloop;



class Channle
{
private:
	// 由外部指明当前描述符什么事件就绪时，执行什么回调函数
	using Event_callback = std::function<void()>;
	int _fd;
	Eventloop * _loop;
	uint32_t _events;					//需要被监控的事件
	uint32_t _revents;	
	Event_callback _read_callback;		//可读事件就绪触发回调函数
	Event_callback _write_callback;		//可写事件就绪触发回调函数
	Event_callback _error_callback;		//错误事件就绪触发回调函数
	Event_callback _close_callback;		//链接断开事件就绪触发回调函数
	Event_callback _any_callback;		//任意事件就绪触发回调函数

public:
	Channle(Eventloop* loop,int fd) 
		:_loop(loop),_fd(fd),_events(0),_revents(0)	
	{
	}
	int FD() {return _fd;}
	// 设置需要监控的事件
	// 这个函数交给Poller，让Poller设置当前触发的事件
	void Set_revents(uint32_t revent) { _revents = revent ;}

	uint32_t Events(){return _events;}

	~Channle()	{}
	void Set_read_callback (const Event_callback& callback) 	{ _read_callback  = callback; }
	void Set_write_callback(const Event_callback& callback)     { _write_callback = callback; }
	void Set_error_callback(const Event_callback& callback) 	{ _error_callback = callback; }
	void Set_close_callback(const Event_callback& callback) 	{ _close_callback = callback; }
	void Set_any_callback  (const Event_callback& callback)     { _any_callback   = callback; }
	// 当前是否监控了描述符是否可读
	bool Readable()	{return (_events & EPOLLIN);}
	// 当前是否监控了描述符是否可写
	bool Writeable(){return (_events & EPOLLOUT);}
	// 监控描述符可读
	void Monitoring_read()
	{
		_events |= EPOLLIN;
		Update();
	}

	// 监控描述符可写
	void Monitoring_write()
	{
		_events |= EPOLLOUT;
		Update();
	}

	// 解除监控可读事件
	void Dismonitoring_read()
	{
		_events &= ~EPOLLIN;
		Update();
	}
	// 解除监控可写事件
	void Dismonitoring_write()
	{
		_events &= ~EPOLLOUT;
		Update();
	}
	// 解除监控所有事件
	void Disable_all()
	{
		_events  = 0;
		Update();
	}
	// 移除监控
	void Remove();
	void Update();
	// 一旦事件触发了，需要调用指定的函数了，就直接调用即可，具体调用哪个函数就在这里面直接判断即可
	void HanlderEvent()
	{
		// 如果符合读事件就绪，那么就进行调用
		if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
		{
			if (_read_callback)_read_callback();
		}
		// 写事件

		if(_revents & EPOLLOUT)
		{
			if(_write_callback) _write_callback();
		}
		else if(_revents & EPOLLERR)
		{
			if(_error_callback) _error_callback();
			
		}
		else if(_revents & EPOLLHUP)
		{
			if(_close_callback) _close_callback();
		}
		if(_any_callback) _any_callback();
	}
};


#define MAX_EPOLLEVENTS 1024
#define MAX_EPOLLSIZE	1024
class Poller
{
private:
	int _epfd;		//表示对应的epoll实例的文件描述符，Linux下一切皆文件
	struct epoll_event _evs[MAX_EPOLLEVENTS];
	std::unordered_map<int, Channle*> _channles;
private:

	// 对Epoll进行操作
	// int epoll_ctl(int __epfd, int __op, int __fd, epoll_event *)
	void Update(Channle* channle ,int op)
	{
		int fd  = channle->FD ();

		// 将需要被监视的FD添加入EPOLL模型之前，我们需要先获取具体的要被监视的动作是什么
		struct epoll_event ev;
		ev.data.fd = fd;
		ev.events  = channle->Events();
		// 将方法填充入epoll_event后，以op即对EPOLL的操作添加进EPOLL实例内部
		int ret = epoll_ctl(_epfd,op,fd,&ev);
		// cout<<"_epfd:"<<_epfd<<endl;
		// cout<<"op"<<op<<endl;
		// cout<<"fd:"<<fd<<endl;
		if(ret < 0)
		{
			ELOG("epoll_ctl Failed! %d",errno);
		}

		return;
	}

	// 判断一个Channle是否被监控
	bool HasChannle(Channle* channle)
	{
		auto it =_channles.find(channle->FD());
		if(it == _channles.end()) return false;
		return true;
	}

public:
	// 对于Poller，我们需要提供三个接口：
	// 1.将对应需要监控的事件和描述符以及执行的回调方法都捆绑到一块
	Poller()
		// :
	{
		_epfd = epoll_create(MAX_EPOLLSIZE);
		if(_epfd < 0)
		{
			ELOG("epoll_create  Failed!");
			abort();
		}
	}

	// 添加和更新监控
	// 不存在就添加，存在就更新
	void Update_events(Channle*channle)
	{
		if(HasChannle(channle) == false)
		{
			_channles.insert(std::make_pair(channle->FD(),channle));
			
			return Update(channle,EPOLL_CTL_ADD);

		}
		return Update(channle,EPOLL_CTL_MOD);
	}

	// 移除对目标的监控
	void Remove_events(Channle* channle)
	{
		auto it =_channles.find(channle->FD());
		if(it != _channles.end())	_channles.erase(it);
		Update(channle,EPOLL_CTL_DEL);
	}

	// 开始监控，返回活跃链接
	// int epoll_wait(int __epfd, epoll_event *__events, int __maxevents, int __timeout)
	void Poll(std::vector<Channle*> *active)
	{
		int nfds = epoll_wait(_epfd,_evs,1024,-1);
		if(nfds < 0)
		{
			if (errno == EINTR) 	return ;
			ELOG("EPOLL WAIT ERROR:%s\n", strerror(errno));
			abort();//退出程序
		}
		for (int i = 0; i < nfds; i++) {
			auto it = _channles.find(_evs[i].data.fd);
			assert(it != _channles.end());
			it->second->Set_revents(_evs[i].events);//设置实际就绪的事件
			active->push_back(it->second);
		}
		return;
	}


};


// 时间轮对象，管理定时器任务对象
// 为了实现延续功能，为两个智能指针定义一个新名称方便编写。
using Runout_Task = function<void()>;
using Release_Task = function<void()>;
class TimeTask
{
private:
	uint64_t _id;				//自身的ID
	uint32_t _timeout; 			//延迟的时间
	bool _cancel;				//决定当前任务对象是否是销毁，true即单纯销毁，false则执行任务
	Runout_Task _runout_task;	//时间到了之后需要执行的任务
	Release_Task _release_task;	//以及释放自身时需要做的事情
public:
	TimeTask(const uint64_t& id,const uint32_t& delay,const Runout_Task& runout)
	:_id(id),_timeout(delay),_runout_task(runout),_cancel(false)
	{}

	void Cancel()	{ _cancel = true ;} 


	// 如何销毁由外部来决定
	void SetReleaseTask(const Release_Task& rt) { _release_task = rt;}

	uint32_t DelayTime(){return _timeout;}
	~TimeTask()
	{
		if(!_cancel)
		_runout_task();
		_release_task();
	}
};

// void  Eventloop::Run_in_loop(const Functor& task);
class Eventloop;

class TimeWheel
{
private:
	using SharedTask = shared_ptr<TimeTask>;
	using WeakTask = weak_ptr<TimeTask>;
	int _tick; 								//滴答指针，指明当前轮转下标
	int _capacity; 								//表盘大小
	vector<vector<SharedTask>> _wheel;			//轮转表盘，设置为二维数组方便扩展任务，比如到时后一次执行多个任务
	unordered_map<uint64_t, WeakTask> _timers;	//便于快速查询获取存放于时间轮内部的定时器任务容器

	int _timer_fd;
	Eventloop* _loop;
	std::unique_ptr<Channle> _channle;

private:
	void ReleaseTask(uint64_t id)
	{
		auto it = _timers.find(id);
		if(it != _timers.end())	_timers.erase(it);
	}

	static int Create_timerfd()
	{
		int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
		if(timerfd < 0)
		{
			ELOG("Timerfd create failed!");
			abort();
		}
		struct itimerspec itime;
		itime.it_value.tv_sec = 1;
		itime.it_value.tv_nsec = 0;					//第一次超时时间为1s后
		itime.it_interval.tv_sec = 1;
		itime.it_interval.tv_nsec = 0;				 //第一次超时后，每次超时的间隔时
		timerfd_settime(timerfd, 0, &itime, NULL);
		return timerfd;
	}

	//获取当前超时了几个任务,也就是timer_fd触发了几次计时
	int Read_timerfd()
	{
		uint64_t readtime;
		int ret = read(_timer_fd,&readtime,sizeof(readtime));
		if(ret < 0)
		{
			ELOG("Timerfd read failed!");
			abort();
		}
		return readtime;

	}

public:
	TimeWheel(Eventloop* loop)
		:_loop(loop),_tick(0),_capacity(60),_wheel(_capacity),_timer_fd(Create_timerfd())
		,_channle(new Channle(_loop, _timer_fd)) 
	{
		// 使用timer_fd来计时读取时间轮，借助Eventloop模块来进行线程运行
		// _timer_行动之后，也就是触发了可读事件，此时就需要处理超时的事件
		_channle->Set_read_callback(std::bind(&TimeWheel :: Ontime,this));

		_channle->Monitoring_read();
	}
	
	// 触发了可读事件就检查一次整体时间轮内部的任务事件
	void Ontime()
	{
		int nums = Read_timerfd();
		for(int i = 0 ; i < nums ; ++i)	TimerWheelRun();
		return;
	}


	// 为了保证线程安全的问题，也就是添加任务的时候很有可能是别的线程添加了一个定时任务，所以我们要将这个定时任务交付给
	// Loop内部的任务队列去执行才是
	void TimerAdd(uint64_t id, int delay, Runout_Task rt);
	// 在IO线程内部执行reflush
	void TimerReflush(uint64_t id);
	
	void TimerCancel(uint64_t id);
	// 提供的函数
	// 添加定时任务,我们只需要直接在该函数内构造对应的TimeTask对象即可
	void Add_task_inloop(uint64_t id, int delay, Runout_Task rt)
	{
		SharedTask pt(new TimeTask (id,delay,rt ));
		// 设置销毁函数，也就是当执行完计时结束后的任务之后，顺便将自身从定时器任务容器内部自己删除，由于this指针的存在
		// 需要使用bind将成员函数传递进去
		pt->SetReleaseTask(bind(&TimeWheel::ReleaseTask,this,id));

		// 存放入任务容器时，不能直接存放shared_ptr，为了防止循环引用我们需要再构造一个weak_ptr存放进去以管理
		_timers[id] = WeakTask(pt);
		// 设置定时时间
		int pos = (_tick + delay) % _capacity;

		_wheel[pos].push_back(pt);

	}

	// 延长定时任务
	void Timer_reflush_inloop(uint64_t id)
	{
		// 延长指定任务
		// 先从容器中取出需要被延长的任务，由于其是weak_ptr 构造而成，我们还要将其转换为shared_ptr
		auto it = _timers.find(id);
		if(it == _timers.end())
		{
			return;
		}
		SharedTask pt = it->second.lock();

		// 新加入一个Shared_ptr进入对应的钟表中
		int delay = pt ->DelayTime();
		int pos = (_tick + delay) % _capacity;
		_wheel[pos].push_back(pt);
	}
	void Timer_cancel_inloop(uint64_t id)
	{
		auto it = _timers.find(id);
		if(it == _timers.end()) return;
		SharedTask pt = it->second.lock();
		if(pt)pt->Cancel();

	}
	//  滴答指针以每一秒走一次的频率释放内部的计时器对象以触发计时器任务生效
	void TimerWheelRun()
	{
		_tick =(_tick + 1) % _capacity;
		_wheel[_tick].clear();
	}

	// 这个接口存在线程安全问题，只能在eventloop内部执行
	bool HasTimer(uint64_t id)
	{
		auto it = _timers.find(id);
		if(it == _timers.end()) return false;
		else return true;
	}
};


// EventLoop模块是前面两个模块的彻底整合，由于是Reactor的设计，可能涉及到多线程执行多任务的问题

// 如果不对上述一个线程有着各自任务的设计去使用EventLoop模块，就会产生线程安全的问题，所以我们需要额外设计一个任务队列
// 所以整体的设计理念如下：
// 1.如果不是当前线程需要管理的文件描述符被捞起，那么就不要直接执行，而是将其压入一个任务队列内等待执行
// 2.如果是当前线程对应的文件描述符，那么就直接执行即可。
// 3.访问任务队列的时候需要一把互斥锁
// 4.设计队列的时候不需要使用队列数据结构,而是使用一个vector来进行模拟,当需要统一处理任务队列的时候,复制一个vector进行执行即可
// 5.需要保存线程ID来检查当前触发的文件描述符事件是否属于当前线程
// 6.当我们以如上的设计进行运行的时候,会存在一个问题,假如待处理事件内部一直没有就绪事件,而其他的线程正在往任务队列内部堆放任务,
// 若此时待处理一直保持阻塞,那么就会导致整个EventLoop被阻塞,所以我们需要一个EVentFD计数器来标定就绪事件的个数

// 上述理解略有偏差，我们设计的是一个线程对应一个循环的设计模式，具体的所有任务执行其实都是服务器使用者指定的。
// 比如监控IO的事件设置由用户决定，而其中的任务队列也是用户添加的，
// 回过头来再理解为什么需要保证线程安全，因为是多reactor模型，用户指定的操作其实就是EventLoop内部需要执行的任务，
class Eventloop
{
private:
	//
	// 血的教训，channle在event_fd 之前初始化了……
	using Functor = std::function<void()>;
	Poller _poller;
	int _event_fd;
	std::mutex _mutex;
	std::thread::id _thread_id;
	std::vector<Functor> _task_queue;
	std::unique_ptr<Channle> _channle;
	TimeWheel _timer_wheel;

public:
	static int Create_eventfd()
	{
		// 
		int retfd = eventfd(0,EFD_CLOEXEC | EFD_NONBLOCK);
		if(retfd < 0)
		{
			ELOG("Eventfd 创建失败！");
			abort();
		}
		return retfd;
	}
	void Run_all()
	{
		// 从任务队列拷贝一份任务队列出来进行执行，尽量不干扰任务队列的运行
		std::vector<Functor> tmp_task;
		{
			std::unique_ptr<std::mutex> _mutex;
			_task_queue.swap(tmp_task);
		}
		for(auto& task : tmp_task) task();
		return;
	}
	// 读取EventFD，也就是清空Efd计数器
	void ReadEventfd()
	{
		uint64_t reader = 0;
		int ret = read(_event_fd,&reader,sizeof(reader));
		if(ret < 0)
		{
			if (errno == EINTR || errno == EAGAIN) return;
			ELOG("Eventfd 读取失败！");
			abort();
		}
		return;
	}
	void WeakUpEventFd()
	{
		uint64_t writer = 0;
		int ret = write(_event_fd,&writer,sizeof(writer));
		if(ret < 0)
		{
			//EINTR -- 被信号打断； EAGAIN -- 表示无数据可读
			if (errno == EINTR ) return;
			ELOG("Eventfd 唤醒失败！");
			abort();
		}
		return;
	}
public:
	// 需要提供的接口有:
	Eventloop()
		:_thread_id(std::this_thread::get_id())
		,_event_fd(Create_eventfd())
		,_channle(new Channle(this, _event_fd))
		,_timer_wheel(this)
	{

		//给eventfd添加可读事件回调函数，读取eventfd事件通知次数
		_channle->Set_read_callback(std::bind(&Eventloop::ReadEventfd,this));

		//启动eventfd的读事件监控
		_channle->Monitoring_read();

	}
	// 设置线程ID
	void Set_threadid(std::thread::id threadid)
	{
		_thread_id = threadid;
	}


	// IO线程来执行这个任务
	void Start()
	{

		while(1)
		{
			std::vector<Channle*> active;
			_poller.Poll(&active);

			for(auto & e : active) e->HanlderEvent();

			Run_all();
		}
	} 


	// 当前分发到Reactor内部的任务是否是属于其内部的IO线程的
	void Run_in_loop(const Functor& task)
	{
		if(Inloop()) return task();
		return Add_task_to_queue(task);
	}

	// 如果不是，那么直接压入任务队列
	void Add_task_to_queue(const Functor& task)
	{
		{
			std::unique_ptr<std::mutex> _mutex;
			_task_queue.push_back(task);
		}
		// 防止EventLoop阻塞在获取fd动静的位置
		WeakUpEventFd();
	}
	// 更新并添加需要监控的描述符对象
	void Update_events(Channle* channle)
	{
		return _poller.Update_events(channle);
	}
	void Remove_events(Channle* channle)
	{
		return _poller.Remove_events(channle);
	}
	// 是否属于当前线程
	bool Inloop()
	{
		return _thread_id == std::this_thread::get_id();
	}
	void Assert_inloop()
	{
		assert(_thread_id == std::this_thread::get_id());
		
	}
	void Add_task_to_timer(uint64_t id,uint32_t delay,const Functor& cb)	{return _timer_wheel.TimerAdd(id,delay,cb);}
	void Reflush_task_in_timer(uint64_t id)									{return	_timer_wheel.TimerReflush(id);}
	void Cancel_task_in_timer(uint64_t id)									{return _timer_wheel.TimerCancel (id);}
	bool HasTimer(uint64_t id) 												{return _timer_wheel.HasTimer(id);}
};

class LoopThread
{
private:
	// 我们需要在整个线程内部创建EventLoop对象才行，不然没办法一一对应一个线程ID
	std:: thread _thread;
	std:: condition_variable _cond; //条件变量
	std:: mutex _mutex;
	Eventloop* _loop;

private:
	// 在线程内部实例化loop对象
	void Loop_entry()
	{
		Eventloop loop;
		{
			unique_lock<mutex> lock(_mutex);
			_loop = &loop;
			_cond.notify_all();
		}
		_loop->Start();
	}
public:
	LoopThread()
		:_loop(nullptr),_thread(std::thread(&LoopThread::Loop_entry,this))
	{
	}

	Eventloop* Get_loop()
	{
		// 获取前，必须要保证_loop不为空
		Eventloop* loop = nullptr;
		{
			unique_lock<mutex> lock(_mutex);
			_cond.wait(lock,[&](){return _loop != nullptr;});
			loop = _loop;
		}

		return loop;
	}


};

class LoopThreadPool
{
private:
	int _thread_nums;
	int _next_idx;
	Eventloop* _baseloop; //主Eventloop，运行于主线程，也就是主reactor
	vector<LoopThread*> _threads;
	vector<Eventloop *> _loops;
public:
	LoopThreadPool(Eventloop* loop )
		:_thread_nums(0),_next_idx(0),_baseloop(loop)
	{}

	void Set_create_nums(int nums)	{  _thread_nums = nums;}

	void Create() //创建所有从属线程
	{
		if(_thread_nums > 0)
		{
			_threads.resize(_thread_nums);
			_loops.resize(_thread_nums);
			for(int i = 0 ; i < _thread_nums ;++i)
			{
				_threads[i] = new LoopThread();
				_loops[i]  = _threads[i]->Get_loop();
 			} 

		return;

		}
	}

	Eventloop* Nextloop() //对分配的loop实行轮转
	{
		if(_thread_nums == 0) return _baseloop;
		int ret = _next_idx % _thread_nums;
		_next_idx++;
		return _loops[ret];
	}

};


// 这个链接对象是处理服务器与客户端之间交互的对象，其中涉及到协议的定制，以及对应阶段的函数
// 数据的发送，数据的读取，缓冲区的使用，对应回调函数的设置。
// 还有作为所有模块的整合，需要用上模块内部的函数
class Connection;
typedef enum  {CONNECTING , CONNECTED, DISCONNECT ,DISCONNECTING} Connect_statu;
using Functor  = std::function<void()>;
using PtrConnection = shared_ptr<Connection>;

class Connection : public enable_shared_from_this<Connection> 
{
private:

	uint64_t _connid; 			 //链接对象的ID，这个ID是唯一的
	int  	 _sockfd;			 //链接对象的fd
	Socket   _socket;			 //套接字管理
	Eventloop* _loop;
	Channle  _channle;			 //连接后的事件管理
	Connect_statu _statu;		 //当前状态Connection的状态,以便继续处理
	bool _active_ontime_destory; //当前是否启用了到时销毁的状态
	Buffer  _in_buffer;			 //输入缓冲区
	Buffer  _out_buffer;		 //输出缓冲区
	Any 	_context;			 //请求连接的上下文内容


	using ConnectedCallback    = std::function<void(const PtrConnection&)>;
	using MessageCallback      = std::function<void(const PtrConnection&,Buffer*)>;
	using CloseCallback 	   = std::function<void(const PtrConnection&)>;
	using AnyEventCallback     = std::function<void(const PtrConnection&)>;
	using ServerCloseCallback  = std::function<void(const PtrConnection&)>;

	
	ConnectedCallback _connect_callback;
	MessageCallback   _message_callback;	
	CloseCallback     _close_callback;
	AnyEventCallback  _event_callback;	


	// 由于connection是被server管理的，我们还需要为其管理者提供一个销毁函数
	ServerCloseCallback     _server_close_callback;

private:
	// Channle的五个回调函数设置
	// 接收socket的数据存放到缓冲区中，然后调用messagecallback
	void Hanlde_read()
	{
		char buf[65536] = {0};
		ssize_t ret = _socket.No_block_recv(buf,65535);
		if(ret  < 0) 	  return Shut_down_inloop(); // 出错了不能直接关闭连接

		// 将获取的数据写入in_buffer
		_in_buffer.Write_push(buf,ret);
		// 调用Messagecallback
		if(_in_buffer.Size_of_read() > 0)
		{
			return _message_callback(shared_from_this(),&_in_buffer);
		}
	}

	// 把发送缓冲区的内容进行发送
	void Hanlde_write()
	{
		// 取出缓冲区内部的数据，然后调用写回调函数
		ssize_t ret = _socket.No_block_send(_out_buffer.Get_read_index(),_out_buffer.Size_of_read());

		if(ret < 0)
		{
			// 发送出错，直接关闭连接
			// 发送之前，还需要检查残余的数据，要把输入缓冲区内部的数据先读完再关闭
			if(_in_buffer.Size_of_read() > 0)
			{
				_message_callback(shared_from_this(),&_in_buffer);
			}
			return  Release_inloop();
		}

		// 正常发送数据，然后移动下标
		_out_buffer.Move_read_index(ret);

		// 如果一轮数据发完了，为了不多次触发loop的写事件，直接关闭对应的写事件监控
		if(_out_buffer.Size_of_read() == 0)
		{
			_channle.Dismonitoring_write();
			// 如果此时正好是准备关闭连接的状态，那么就直接关闭连接
			if(_statu==DISCONNECTING)
			return  Release_inloop();
		}

		return;
		
	}
	void Hanlde_close()
	{
			if(_in_buffer.Size_of_read() > 0)
			{
				_message_callback(shared_from_this(),&_in_buffer);
			}
			return  Release_inloop();
	}
	void Hanlde_error()
	{
		return Hanlde_close();
	}
	void Hanlde_any()
	{
		if(_active_ontime_destory == true) _loop->Reflush_task_in_timer(_connid);
		if(_event_callback) _event_callback(shared_from_this());
	}


	void Send_inloop(Buffer& buf)	//将数据发送至缓冲区，启动写监控（?）
	{
		// 如果当前状态是等待关闭连接的状态，则不发送数据
		if(_statu == DISCONNECTING) return;	
		_out_buffer.Write_as_buffer_push(buf);
		if(_channle.Writeable() == false) _channle.Monitoring_write();

	}

	void Established_inloop()			//连接获取之后所处的状态下需要进行的各种设置（给channle设置读回调，启动读监控）
	{
		// 1.修改连接状态
		assert(_statu ==CONNECTING);
		_statu = CONNECTED;
		// 2.启动读事件监控
		_channle.Monitoring_read();
		// 3.调用回调函数
		if(_connect_callback) _connect_callback(shared_from_this());
	}


	//实际的释放接口
	void Release_inloop()
	{
		// 1.修改连接状态
		_statu =  DISCONNECT;
		// 2.移除描述符的事件监控
		_channle.Remove();
		// 3.关闭描述符
		_socket.Close();

		// 4.如果计时器内部还有销毁任务，则取消任务

		if(_loop->HasTimer(_connid))
		_loop->Cancel_task_in_timer(_connid);

		// 5.调用关闭回调函数,用户设置的关闭回调函数
		// 避免先移除了服务器的管理信息，导致的Connection被先释放的问题
		if(_close_callback) _close_callback(shared_from_this());

		// 6.清除服务器内部的信息，也就是调用服务器的回调函数
		if(_server_close_callback) _server_close_callback(shared_from_this());

	}		

	//并不直接关闭连接，还需要额外处理缓冲区内部的数据
	// 
	void Shut_down_inloop()		
	{
		_statu = DISCONNECTING;// 设置连接为半关闭状态
		if (_in_buffer.Size_of_read() > 0) {
			if (_message_callback) _message_callback(shared_from_this(),&_in_buffer);
		}
		//要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
		if (_out_buffer.Size_of_read() > 0) {
			if (_channle.Writeable() == false) 	_channle.Writeable();
		}
		if (_out_buffer.Size_of_read() == 0) {
		Release_inloop();
		}
	} 

	void Enable_ontime_destory_inloop(int sec) // 启动定时销毁
	{
		// 1.将定时销毁状态设置为true
		_active_ontime_destory = true;

		// 2.添加定时销毁任务
		if(_loop->HasTimer(_connid) == false)
		{
			_loop->Add_task_to_timer(_connid,sec,std::bind(&Connection::Release_inloop,this));
		}
		return _loop->Reflush_task_in_timer(_connid);
	}

	void Cancel_ontime_destory_inloop() 	//关闭定时销毁
	{
		_active_ontime_destory = false;
		if(_loop->HasTimer(_connid) != false)
		{
			_loop->Cancel_task_in_timer(_connid);
		}

	}
	void Upgrade_inloop(Any context, const ConnectedCallback& conncb, const MessageCallback& messcb,
				 const CloseCallback& closecb,const AnyEventCallback& eventcb)
	{
			_context = context;
			_connect_callback = conncb;
			_message_callback = messcb;	
			_close_callback   = closecb;
			_event_callback	  = eventcb;	
	}
public:

	// 如下是成员变量的设置与获取
	Connection(Eventloop* loop,uint64_t id,int socketid)
		:_connid(id),_sockfd(socketid),_active_ontime_destory(false),_loop(loop),_statu(CONNECTING)
		,_socket(_sockfd),_channle(loop,_sockfd)

	{
		_channle.Set_read_callback  (std::bind(&Connection::Hanlde_read,this ));
		_channle.Set_write_callback (std::bind(&Connection::Hanlde_write,this));
		_channle.Set_close_callback (std::bind(&Connection::Hanlde_close,this));
		_channle.Set_error_callback (std::bind(&Connection::Hanlde_error,this));
		_channle.Set_any_callback	(std::bind(&Connection::Hanlde_any,this  ));

	}

	int FD()	  		{return _sockfd;}
	uint64_t ID() 		{return _connid;}
	Any* Get_context()	{return &_context;}
	void Set_context(const Any& context)	{_context = context;}
	~Connection()	{ELOG("析构一个Connection,地址：%p",this);}
public:
	//连接阶段的回调函数设置 
	void Set_Connected_callback  (const ConnectedCallback& cb) 	{ _connect_callback = cb;}
	void Set_Message_callback    (const MessageCallback& cb) 	{ _message_callback = cb;}
	void Set_Close_callback	     (const CloseCallback& cb) 		{ _close_callback   = cb;}
	void Set_Anyevent_callback   (const AnyEventCallback& cb) 	{ _event_callback   = cb;}
	void Set_sever_close_callback(const ServerCloseCallback& cb){_server_close_callback = cb;}


public:
	//向外部提供的接口
	void Established()
	{
		_loop->Run_in_loop(std::bind(&Connection::Established_inloop,this));
	}

	void Send(const char* data,size_t len)	//将数据发送至缓冲区，启动写监控
	{
		// 但是在这里有一个问题，我们设计的服务器执行任务的时候，是将任务压入了任务池，所以，send的这个任务很有可能不是立即执行的
		// 这样就有可能当任务线程执行这个任务的时候，data可能已经被释放了，所以我们修改一下函数让其保持一个临时生命周期
		Buffer buf;
		buf.Write_push(data ,len);
		_loop->Run_in_loop(std::bind(&Connection::Send_inloop,this,buf));

	}

		void Release()
	{
		_loop->Add_task_to_queue(std::bind(&Connection::Release_inloop, this));
	}

	//这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
	void Shut_down()		//并不直接关闭连接，还需要额外处理缓冲区内部的数据
	{
		_loop->Run_in_loop(std::bind(&Connection::Shut_down_inloop,this));
	} 

	void Enable_ontime_destory(int sec)
	{
		_loop->Run_in_loop(std::bind(&Connection::Enable_ontime_destory_inloop,this,sec));
	}

	void Cancel_ontime_destory() 	//关闭定时销毁
	{
		_loop->Run_in_loop(std::bind(&Connection::Cancel_ontime_destory_inloop,this));

	}

	// 更新协议，也需要更新回调函数
	void Upgrade(Any context, const ConnectedCallback& conncb, const MessageCallback& messcb,
				 const CloseCallback& closecb,const AnyEventCallback& eventcb)
	{
		_loop->Assert_inloop();
		_loop->Run_in_loop(bind(&Connection::Upgrade_inloop,this,context,conncb,messcb,closecb,eventcb));
	}



};


// 只监控监听套接字，具体的回调函数设置还需要服务器模块来设置。
class Acceptor
{
private:
using Acceptor_callback = std::function<void(int)>;

	Socket _socket;
	Eventloop* _loop;
	Channle _channle;
	Acceptor_callback  _accept_cb;
private:
	void HanldeRead()
	{
		DLOG("Acceptor已触发可读！");

		int newfd = _socket.Accept();
		if(newfd < 0 ) return;

		if(_accept_cb) _accept_cb(newfd);
	}

	int CreatServer(int port)
	{
		bool ret = _socket.CreatServer(port);
		assert(ret==true);
		return _socket.FD();
	}

public:
	Acceptor(Eventloop* loop,int port)
		:_loop(loop),_socket(CreatServer(port)),_channle(_loop,_socket.FD())
	{
		
		_channle.Set_read_callback(std::bind(&Acceptor::HanldeRead,this));
	}

	void Start_listen()
	{
		_channle.Monitoring_read();
	}

	void Set_accept_callback(const Acceptor_callback& cb )
	{
		_accept_cb = cb;
	}

};


// 作为所有的模块整合的类，主体的功能就是管理连接，指定线程池大小，管理监听套接字，为用户提供设置回调函数的接口
// 还有定时删除，定时任务等等等等
class TcpServer
{
private:
	uint64_t _nextid;  				//连接的ID
	int _port;
	int _timeout;						//多久之后销毁当前Connection
	bool _enable_ontime_destory;		//是否启动定时销毁
	Eventloop _baseloop;				//主线程的reactor

	Acceptor _acceptor;					//接收器模块
	LoopThreadPool _threadpool;
	unordered_map<uint64_t, PtrConnection> _conns;

	using ConnectedCallback = std::function<void(const PtrConnection&)>;
	using MessageCallback   = std::function<void(const PtrConnection&,Buffer*)>;
	using CloseCallback 	= std::function<void(const PtrConnection&)>;
	using AnyEventCallback  = std::function<void(const PtrConnection&)>;
	
	ConnectedCallback _connect_callback;
	MessageCallback   _message_callback;	
	CloseCallback     _close_callback;
	AnyEventCallback  _event_callback;	
private:
	//为新连接构造一个新Connection对象
	void Create_new_connection(int fd)
	{
		_nextid++;
		PtrConnection conn( new Connection(_threadpool.Nextloop(),_nextid,fd));
		// // 在这里，已经面向了新链接的套接字进行管理，这个时候为其添加回函数
		// 此处对connection对象的连接阶段设置回调函数，设置的参数已经在内部传入了参数，所以我们在绑定的时候指定一个placeholder即可
		conn->Set_Connected_callback	(_connect_callback);
		conn->Set_Message_callback		(_message_callback);
		conn->Set_Close_callback		(_close_callback);
		conn->Set_Anyevent_callback 	(_event_callback);
		conn->Set_sever_close_callback  (bind(&TcpServer::Remove_connection,this,placeholders::_1));
		// 启动超时销毁

		if(_enable_ontime_destory)	conn->Enable_ontime_destory(_timeout);
		
		conn->Established();
		_conns.insert(make_pair(_nextid,conn));
		DLOG("获取新连接");
	}

	void Remove_connection_inloop(const PtrConnection& conn)
	{
		int id  = conn->ID();
		auto it = _conns.find(id);
		if(it != _conns.end())
		{
			_conns.erase(id);
		}
	}

	//移除某个Connection
	void Remove_connection(const PtrConnection& conn)
	{
		_baseloop.Run_in_loop(bind(&TcpServer::Remove_connection_inloop,this,conn));
	}

		// void Add_task_to_timer(uint64_t id,uint32_t delay,const Functor& cb)
	void RunAfter_inloop(const Functor& task,int delay)
	{
		_nextid++;
		_baseloop.Add_task_to_timer(_nextid,delay,task);
	}

public:
	TcpServer(int port)
		:_port(port),_nextid(0),_enable_ontime_destory(false),_acceptor(&_baseloop,port),_threadpool(&_baseloop)
	{
		// 在构造函数阶段，启动监听套接字的监听
		_acceptor.Set_accept_callback(bind(&TcpServer::Create_new_connection,this,placeholders::_1));
		_acceptor.Start_listen();
	}
	// 启动服务器运转，启动监听
	void Start()
	{
		_threadpool.Create();
		return _baseloop.Start();
	}
	//提供一个设置线程池数量的接口的函数
	void Set_threadnums(int n)		{_threadpool.Set_create_nums(n);}
	// 用户回调函数设置接口
	void Set_Connected_callback(const ConnectedCallback& cb) 	{ _connect_callback = cb;}
	void Set_Message_callback  (const MessageCallback& cb) 		{ _message_callback = cb;}
	void Set_Close_callback	   (const CloseCallback& cb) 		{ _close_callback   = cb;}
	void Set_Anyevent_callback (const AnyEventCallback& cb) 	{ _event_callback   = cb;}
	
	void Enable_ontime_destory(int time)
	{
		_timeout = time;
		_enable_ontime_destory = true;
	}

	// 添加一个定时任务
	void RunAfter(const Functor& task,int delay)
	{
		_baseloop.Run_in_loop(bind(&TcpServer::RunAfter_inloop,this,task,delay));
	}


};
	
	
	// 移除监控
	void Channle::Remove()
	{
		_loop->Remove_events(this);
	}
	void Channle::Update()
	{
		_loop->Update_events(this);
	}
	// 在IO线程内部执行reflush
	void TimeWheel::TimerAdd(uint64_t id, int delay, Runout_Task rt)
	{
		_loop->Run_in_loop(std::bind(&TimeWheel::Add_task_inloop,this,id,delay,rt)); 
	}
	void TimeWheel::TimerReflush(uint64_t id)
	{
		_loop->Run_in_loop(std::bind(&TimeWheel::Timer_reflush_inloop, this,id));
	}
	void TimeWheel::TimerCancel(uint64_t id)
	{
		_loop->Run_in_loop(std::bind(&TimeWheel::Timer_cancel_inloop, this,id));

	}

class NetWork
{
	public:
	NetWork()
	{
		signal(SIGPIPE,SIG_IGN);
	}
};
static NetWork nw;